package com.talent.service.iot.mqtt.service;

import com.talent.service.iot.mqtt.model.TlMqttSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @ClassName：DefaultMqttSessionServiceImpl
 * @Author: hszhou
 * @Date: 2025/1/8 9:11
 * @Description: 默认的session处理实现类 使用本地
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.LOCAL)
public class LocalMqttSessionServiceImpl implements IMqttSessionService{

    private static final Logger logger = LoggerFactory.getLogger(LocalMqttSessionServiceImpl.class);

    /**
     * 对应的是客户端的ID后面是客户端对应的session
     */
    public static final ConcurrentHashMap<String, TlMqttSession> CLIENT_TO_SESSION=new ConcurrentHashMap<String, TlMqttSession>();

    @Override
    public Mono<TlMqttSession> save(TlMqttSession session) {
        String clientId = session.getClientId();
        CLIENT_TO_SESSION.put(clientId,session);
        return Mono.justOrEmpty(session);
    }

    @Override
    public  Mono<TlMqttSession> find(String clientId) {
        TlMqttSession tlMqttSession = CLIENT_TO_SESSION.get(clientId);
        return Mono.justOrEmpty(tlMqttSession);
    }

    @Override
    public  Mono<Boolean> clear(String clientId) {
        TlMqttSession remove = CLIENT_TO_SESSION.remove(clientId);
        if (remove!=null){
            return Mono.justOrEmpty( true);
        }
        return Mono.justOrEmpty( false);

    }

    @Override
    public Boolean hasKey(String clientId) {
        return CLIENT_TO_SESSION.get(clientId)!=null;
    }

    @Override
    public Flux<String> clients() {
        ConcurrentHashMap.KeySetView<String, TlMqttSession> keys = CLIENT_TO_SESSION.keySet();
        logger.info("当前链接的客户端数为【{}】",keys.size());
        return Flux.fromIterable(keys);
    }
}
